package com.all.controller;

import com.all.entity.User;
import com.all.service.IUserService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitter;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import org.springframework.web.servlet.mvc.method.annotation.StreamingResponseBody;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Date;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;

/**
 * @Author LY
 * @create 2024/7/12 16:28
 * @Description 用户层 异步模式（ResponseBodyEmitter、SseEmitter、StreamingResponseBody）
 */
@RestController
@RequestMapping("/user")
public class UserController {

    private final CopyOnWriteArrayList<SseEmitter> emitters = new CopyOnWriteArrayList<>();

    @Autowired
    private IUserService iUserService;

    /**
     * SseEmitter 实现异步传输
     */
    @GetMapping(value = "/get",produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public  SseEmitter getAllUsers(){
        SseEmitter emitter = new SseEmitter();
        this.emitters.add(emitter);

        emitter.onCompletion(() -> this.emitters.remove(emitter));
        emitter.onError((e) -> this.emitters.remove(emitter));
        emitter.onTimeout(() -> this.emitters.remove(emitter));

        return emitter;
    }

    /**
     * 向SseEmitter发送添加的用户
     */
    @GetMapping("/add")
    public void addUser(){
        User user = new User();
        LocalDateTime now = LocalDateTime.now();
        user.setUserId(System.currentTimeMillis());
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        user.setUserName(formatter.format(now));
        user.setPassword(UUID.randomUUID().toString());
        user.setEmail("admin@all.com");
        user.setAddress("广东深圳");
        user.setPhone("15100000000");
        iUserService.addUser(user);
        sendToClients();
    }

    public void sendToClients(){
        List<User> users = iUserService.getUsers();
        for (SseEmitter emitter : emitters) {
            try {
                emitter.send(users);
            } catch (Exception e) {
                emitter.completeWithError(e);
            }
        }
    }

    /**
     * ResponseBodyEmitter 实现异步传输
     */
    @GetMapping("/bodyEmitter")
    public ResponseEntity<ResponseBodyEmitter> handler(){
        //  创建一个ResponseBodyEmitter，-1代表不超时
        ResponseBodyEmitter emitter = new ResponseBodyEmitter(-1L);
        //  异步执行耗时操作
        CompletableFuture.runAsync(() -> {
            try {
                //  模拟耗时操作
                for (int i = 0; i < 1000; i++){
                    System.out.println("bodyEmitter:" + i);
                    //  发送数据
                    emitter.send("bodyEmitter " + i + "@ " + new Date() + "\n");
                    Thread.sleep(2000);
                }
                //  完成
                emitter.complete();
            }catch (Exception e){
                //  异常结束接口
                emitter.completeWithError(e);
            }
        });
        return ResponseEntity.ok().contentType(MediaType.TEXT_HTML).body(emitter);
    }

    /**
     * StreamingResponseBody 实现异步传输
     */
    @GetMapping("/stream2")
    public ResponseEntity<StreamingResponseBody> stream() {
        StreamingResponseBody stream = out -> {
            String message = "fyk_test_";
            for (int i = 0; i < message.length(); i++) {
                try {
                    out.write((message + i).getBytes());
                    out.write("\r\n".getBytes());
                    //  调用一次flush就会像前端写入一次数据
                    out.flush();
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
        return ResponseEntity.ok().contentType(MediaType.TEXT_HTML).body(stream);
    }

}
